package yuekao3.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import yuekao3.entity.Tm5;
import yuekao3.entity.Tm6;
import yuekao3.util.ClickHouseUtil;
import yuekao3.util.KafkaUtil;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.HashSet;

public class Statistics {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //（DWS层）编写Flink DataStream程序，实时消费DWD宽表数据，进行每日下单数据实时统计，按照省份、商品维度分组，统计销售订单总数、销售额和商品数量：（8分）
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSources("dwd_db_orders"));
//        streamSource.print();
        SingleOutputStreamOperator<Tm5> operator = streamSource.map(new MapFunction<String, Tm5>() {
            @Override
            public Tm5 map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String id = jsonObject.getJSONObject("f0").getString("id");
                String province_id = jsonObject.getJSONObject("f0").getString("province_id");
                String name = jsonObject.getJSONObject("f3").getString("name");
                String sku_id = jsonObject.getJSONObject("f1").getString("sku_id");
                String sku_name = jsonObject.getJSONObject("f4").getString("sku_name");
                Integer sku_num = jsonObject.getJSONObject("f1").getInteger("sku_num");
                Double order_price = jsonObject.getJSONObject("f1").getDouble("order_price");
                String create_time = jsonObject.getJSONObject("f0").getString("create_time");
                return new Tm5(id, province_id, name, sku_id, sku_name, sku_num, order_price, create_time);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<Tm5>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> {
                    try {
                        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));
//        operator.print();
        //6）、（DWS层）自定义窗口延迟关闭时间，及延迟数据侧边流输出，保存Kafka消息队列和同时写入数据库表中；（7分）
        //进行每日下单数据实时统计，按照省份、商品维度分组，统计销售订单总数、销售额和商品数量：（8分）
        OutputTag<Tm5> outputTag = new OutputTag<Tm5>("outputTag") {
        };
        SingleOutputStreamOperator<Tm6> process = operator.keyBy(new KeySelector<Tm5, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(Tm5 one) throws Exception {
                        return new Tuple4<>(one.getProvince_id(), one.getName(), one.getSku_id(), one.getSku_name());
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(1)))
                .sideOutputLateData(outputTag)
                .process(new ProcessWindowFunction<Tm5, Tm6, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple4<String, String, String, String> one,
                                        ProcessWindowFunction<Tm5, Tm6, Tuple4<String, String, String, String>, TimeWindow>.Context context,
                                        Iterable<Tm5> iterable,
                                        Collector<Tm6> collector) throws Exception {
                        HashSet<String> count = new HashSet<>();
                        Double sumprice = 0.0;
                        Integer sum = 0;
                        for (Tm5 o : iterable) {
                            count.add(o.getId());
                            sumprice += o.getOrder_price();
                            sum += o.getSku_num();
                        }
                        String eet = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(context.window().getStart());
                        String ent = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(context.window().getEnd());
                        collector.collect(new Tm6(eet, ent, count.size(), sumprice, sum));
                    }
                });
//        process.getSideOutput(outputTag).print("outputTag:");
//        process.print();
        //7）、（DWS层）clickhouse数据库创建表，保存下单汇总聚合宽表数据，采用upsert方式，主键不存在插入数据，主键相同时更新数据；（7分）
        process.addSink(new ClickHouseUtil());

        env.execute();
    }
}
